package com.atguigu.Flink.timeAndwindow;

import com.atguigu.Flink.POJO.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class Flink03_WaterMarkSend {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",6789);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        DataStreamSource<String> ds1 = env.socketTextStream("hadoop102", 8888);
        DataStreamSource<String> ds2 = env.socketTextStream("hadoop102", 9999);
        SingleOutputStreamOperator<Event> eventSingleOutputStreamOperator = ds1.map(
                line -> {
                    String[] split = line.split(",");
                    return new Event(split[0], split[1], Long.valueOf(split[2]));
                }
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event event, long l) {
                        return event.getTs();
                    }
                })
        );
        //eventSingleOutputStreamOperator.map(x -> x).setParallelism(2).print();

        SingleOutputStreamOperator<Event> eventSingleOutputStreamOperator1 = ds2.map(
                line -> {
                    String[] split = line.split(",");
                    return new Event(split[0], split[1], Long.valueOf(split[2]));
                }
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event event, long l) {
                        return event.getTs();
                    }
                })
        );
        eventSingleOutputStreamOperator.union(eventSingleOutputStreamOperator1).map(x->x).print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
